package kafkaStreaming

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.util
//import org.apache.spark.sql.execution.streaming.CommitMetadata.format

object KafkaDstreamNew {


  // countState


  // 2
  case class stuInfo(sex:String,num:Int)
  def main(args: Array[String]): Unit = {

  val conf = new SparkConf().setMaster("local[*]").setAppName("kafkaDstream")
  val ssc = new StreamingContext(conf,   Seconds(5)  )

    // 1) 导入 sparksql 的依赖
  val  sqlCon = new SQLContext(ssc.sparkContext)

    ssc.sparkContext.setLogLevel("error")

   // kafka 配置
    // 线上 123.56.187.176   cheng

 // 1） kafka 依赖   2） sparkstreaming-kafka
 // 方式1： jas 文件夹  ，放在 jas文件夹里边  ，很多的 依赖的jar包
//  方式2： pom 导入

   val topic =  "stuInfo"
   val group = "niit1212"

    val kafkaParams = Map[String, Object](

   // 1  kafka 所在地址    cheng   123.56.187.176
   "bootstrap.servers" -> "cheng:9092",

   "key.deserializer" -> classOf[StringDeserializer],
   "value.deserializer" -> classOf[StringDeserializer],

   "group.id" -> group,

   // 消费模式  从头消费， 从尾消费，当前消费
   "auto.offset.reset" -> "earliest",

   // 消费的元数据 信息，  消费的位置
   // 是否自动提交
   "enable.auto.commit" -> (false: java.lang.Boolean)
 )

    ssc.checkpoint("./checkpoint")

    // 连接kafka
   val linesStream =   KafkaUtils.createDirectStream(
       ssc,
       // 策略     PreferConsistent   kafka 集群 master /leader
       PreferConsistent,

       ConsumerStrategies.Subscribe[String, String](Array(topic), kafkaParams)
     )

    // linesStream   key 偏移量   value 值

    // linesStream value 真正的数据
    val line = linesStream.map(    _.value()      )

    //  line  每一行数据
    // 逻辑

    //  line     Dstream       ----》  seq [ RDD ]
    // RDD       ----》  Array  [( 1,3) ,( 2,3)    ]
    // RDD   ----》 ( 1,3)  tuple

    //  1,3
    //  2,4


    line.foreachRDD(
          x=>{
            // stuInfo
            // sex     3

            //  x   rdd
           val result =   x.map(
              line =>{
                // sex
                val  sex = line.split("\t")(2)
                var  x =  ""
                if (sex == "1"){
                  // 1 性别  1 映射的1  (key,1)
                  x = sex
                }else if( sex == "0"){
                  // 0  性别   ，1   映射的1  (key,1)
                  x = sex
                }
                (x,1)
              }
            ).reduceByKey(_+_)
            //  reduceByKey      RDD [("java",10) ,  ("spark",11) ]

            //  result      RDD [("java",10) ,  ("spark",11) ]


            //   结果   result   RDD [("java",10) ,  ("spark",11) ]
            //               ("java",10)  tuple
            result.foreach(
              y => {
                // y    ("java",10)
               val sex =  y._1
               val num = y._2

               // res
               val res = sex+"," + num

              // 发送 kafka

         val kafkapro = new util.HashMap[String, Object]()


          kafkapro.put("bootstrap.servers", "cheng:9092")
         kafkapro.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
          kafkapro.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")


           val producer = new KafkaProducer[String, String](kafkapro)
                producer.send(new ProducerRecord[String, String]("t11111", res))
        producer.close()













              }


            )


          }


     )


    // 监听器
    ssc.start()
    ssc.awaitTermination()



  }


}
